package com.leilei;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.RichAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.stream.StreamSupport;

/**
 * @author lei
 * @version 1.0
 * @date 2021/4/11 13:47
 * @desc 窗口算子
 */
public class WindowOperator {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setParallelism(1);
        // 从 1-10 加载数据 ，共十个元素
        DataStreamSource<Long> source = env.fromSequence(1, 10);
        // 每两个元素 计算一次结果
        AllWindowedStream<Long, GlobalWindow> windowedStream = source.countWindowAll(2);
        SingleOutputStreamOperator<Long> result = windowedStream.apply(new RichAllWindowFunction<Long, Long, GlobalWindow>() {
            @Override
            public void apply(GlobalWindow window, Iterable<Long> values, Collector<Long> out) {
                Long sum = StreamSupport.stream(values.spliterator(), false)
                        .reduce(Long::sum).orElse(99999L);
                out.collect(sum);

            }
        });
        result.print("window");
        env.execute();
    }
}
